package cn.shiva.redisunit.message.stream;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import java.time.Duration;

/**
 * @author shiva   2022-08-16 21:12
 */
@Configuration
public class MessageStreamConfig {

    @Autowired
    private MessageStreamListener streamListener;
    @Autowired
    private RedisTemplate redisTemplate;

    @Bean
    public Subscription subscription(RedisConnectionFactory factory) {
        EventStreamUtils.createConsumerGroup("mystream", "mygroup", redisTemplate);
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(1))
                .build();
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> listenerContainer = StreamMessageListenerContainer.create(factory, options);
        Subscription subscription = listenerContainer.receiveAutoAck(Consumer.from("mygroup", "huhailong"),
                StreamOffset.create("mystream", ReadOffset.lastConsumed()), streamListener);
        listenerContainer.start();
        return subscription;
    }

}
